package com.majun;

import java.util.Arrays;
import java.util.concurrent.atomic.AtomicInteger;

import cn.hutool.core.util.StrUtil;
import reactor.core.publisher.Flux;

public class ReactorApp {

	public static void main(String[] args) {
		streamTest();
		System.err.println("===========");
		reactorTest();
	}

	public static void streamTest() {
		String[] WORDS = { "the", "quick", "brown", "fox", "jumped", "over", "the", "lazy", "dog" };
		AtomicInteger index = new AtomicInteger(1);
		Arrays.stream(WORDS).map(word -> StrUtil.format("{}. {}", index.getAndIncrement(), word))
				.forEach(System.out::println);
	}

	public static void reactorTest() {

		// 序号① 的代码 Flux 是我们之前提到的 一个能够发出 0 到 N 个元素的响应流发布者，fromArray 是它的静态方法，用来创建 Flux 响应流
		// 序号② 的代码 Flux 的 range 操作符和 Stream 的 range 相同，用来生成 整数 Flux 响应流；zipWith
		// 操作符用来合并两个 Flux，并将响应流中的元素一一对应，当其中一个响应流完成时，合并结束，之前未结束的响应流剩下的元素将被忽略
		// 序号③ 的代码 zipWith 操作符 支持传递一个 BiFunction
		// 的函数式接口实现，定义如何来合并两个数据流中的元素，本例中我们将索引和单词连接起来
		// 序号④ 的代码 subscribe 即为订阅方法，此处我们做了数据流中元素输出至控制台
		String[] WORDS = { "the", "quick", "brown", "fox", "jumped", "over", "the", "lazy", "dog" };
		Flux.fromArray(WORDS) // ①
				.zipWith(Flux.range(1, Integer.MAX_VALUE), // ②
						(word, index) -> StrUtil.format("{}. {}", index, word)) // ③
				.subscribe(System.err::println); // ④

	}
}
